package com.intct.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-11-13 16:14
 */
public class DataGenTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql("CREATE TABLE source (" +
                "    id INT," +
                "    ts BIGINT," +
                "    vc INT" +
                ") WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.id.kind'='random'," +
                "    'fields.id.min'='1'," +
                "    'fields.id.max'='10'," +
                "    'fields.ts.kind'='sequence'," +
                "    'fields.ts.start'='1'," +
                "    'fields.ts.end'='1000000'," +
                "    'fields.vc.kind'='random'," +
                "    'fields.vc.min'='1'," +
                "    'fields.vc.max'='100'" +
                ")");

        tenv.executeSql("WITH source_with_total AS (" +
                "    SELECT id, vc+10 AS total" +
                "    FROM source" +
                ")" +
                " SELECT id, SUM(total)" +
                " FROM source_with_total" +
                " GROUP BY id ").print();

//        tenv.executeSql("CREATE TABLE sink (" +
//                "    id INT," +
//                "    ts BIGINT," +
//                "    vc INT" +
//                ") WITH (" +
//                "'connector' = 'print'" +
//                ")");
    }
}
